﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using PServiceBus.Gateway.Interfaces.DTO;
using System.Collections.Concurrent;
using System.Threading;
using PServiceBus.Services.Gateway.Extensions;

namespace PServiceBus.Services.Gateway.Runtime {
    internal sealed class TopicBatch {

        const int MINIMUM_CORE = 4,
                  MINIMUM_MULTIPLY = 2,
                  MINIMUM_BATCH_SIZE = 1000,
                  MINUMUM_INTERVAL = 100,
                  MAX_BATCH_SIZE = MINIMUM_BATCH_SIZE * MINUMUM_INTERVAL;
        
        private static readonly int RATE_OFFSET = (int)Math.Ceiling(Environment.ProcessorCount / (MINIMUM_CORE * 1.0));

        private static readonly int BATCH_SIZE = MINIMUM_BATCH_SIZE * RATE_OFFSET;

        //Decrease interval on cpu with more than minimum_core to allow better throughput
        private static readonly int INTERVAL_MS = 
            MINUMUM_INTERVAL - (Environment.ProcessorCount > MINIMUM_CORE ? Environment.ProcessorCount : 0);

        private static ConcurrentDictionary<string, ConcurrentQueue<Topic>> _topics =
            new ConcurrentDictionary<string, ConcurrentQueue<Topic>>();

        private static ConcurrentDictionary<string, RegisteredWaitHandle> _handles =
            new ConcurrentDictionary<string, RegisteredWaitHandle>();
        private static ConcurrentDictionary<string, ManualResetEvent> _resets =
            new ConcurrentDictionary<string, ManualResetEvent>();

        static TopicBatch() {
            ESB.OnDisconnect = Cleanup;
        }

        private static void Cleanup() {
            foreach (var reset in _resets)
                reset.Value.Set();
        }

        private static Topic GetNextTopic(ConcurrentQueue<Topic> queue) {
            Topic topic;
            if (queue.TryDequeue(out topic))
                return topic;
            return null;
        }

        private static void ProcessTopic(string name) {
            var queue = _topics[name];

            var topics = new List<Topic>();

            for (var i = 0; i < BATCH_SIZE; i++) {
                var tp = GetNextTopic(queue);
                if (tp != null)
                    topics.Add(tp);
            }

            if (topics.Count == 0) return;

            var messages = new List<Dictionary<string, object>>();
            var topicCount = topics.Count;
            for (var i = 0; i < topicCount; i++) {
                var pubDicts = topics[i].PublishDicts;
                var pubDictCount = pubDicts.Count;
                for (var j = 0; j < pubDictCount; j++) {
                    messages.Add(pubDicts[j]);
                }
            }


            var topic = topics[0].ToTopicInfo();
            if (messages.Count > MAX_BATCH_SIZE)
                ProcessTopicAsChunk(topic, messages);
            else {
                topic.PublishDicts = messages;
                ESB.Execute(messageBus => messageBus.PublishTopic(topic));
            }
            
            messages = null;
        }

        private static void ProcessTopicAsChunk(TopicInfo topic, List<Dictionary<string, object>> messages) {
            var messageSize = messages.Count;
            var batchSize = BATCH_SIZE * MINIMUM_MULTIPLY;
            var size = (int)Math.Floor(messageSize / (decimal)batchSize);
            var diff = messageSize - ((size - 1) * batchSize);
            var skip = 0;

            for (var i = 0; i < size; i++) {
                var count = (i + 1) * batchSize;
                count = count > messageSize ? diff : batchSize;
                topic.PublishDicts = messages.GetRange(skip, count);
                ESB.Execute(messageBus => messageBus.PublishTopic(topic));
                skip += count;
            }
        }

        private static RegisteredWaitHandle CreateHandler(string name) {
            return ThreadPool.RegisterWaitForSingleObject(
                    _resets.GetOrAdd(name, _ => new ManualResetEvent(false)),
                    new WaitOrTimerCallback(Process),
                    name, INTERVAL_MS, true);
        }

        private static void Process(object state, bool timeout) {
            var name = state as string;
            if (timeout) {
                try { ProcessTopic(name); } catch{ }
                _handles[name] = ThreadPool.RegisterWaitForSingleObject(_resets[name],
                    new WaitOrTimerCallback(Process), name, INTERVAL_MS, true);
            } else {
                RegisteredWaitHandle handle;
                if (_handles.TryRemove(name, out handle))
                    handle.Unregister(null);
            }
        }

        private static void CreateProcessor(string name) {
            if (_handles.ContainsKey(name)) return;
            _handles[name] = CreateHandler(name);
        }

        internal static void Publish(Topic topic){
            var name = topic.TopicName;
            _topics.GetOrAdd(name, _ => new ConcurrentQueue<Topic>()).Enqueue(topic);
            CreateProcessor(name);
        }
    }
}
